/*
 * Author:   W.SH
 */
package com.wgg.crawler.scheduler;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

import com.wgg.crawler.filter.BloomFilter;
import com.wgg.crawler.util.Config;

/**
 * 〈一句话功能简述〉<br> 
 * 〈功能详细描述〉
 *
 *  @author W.SH
 */
public class BloomScheduler extends SchedulerMonitor implements Scheduler {
    
    private static final Logger LOG = Logger.getLogger(BloomScheduler.class.getName());
    
    private  final Object lock = new Object();
    
    private BlockingQueue<String> queue = new LinkedBlockingDeque<String>();
    private BloomFilter<String> bloomFilter = new BloomFilter<String>(Config.BLOOMFILTER_ERROR_RATE, Config.BLOOMFILTER_RATE); 

    @Override
    public String poll() throws InterruptedException {
        String url = queue.poll(2000, TimeUnit.MILLISECONDS);
        if (null != url) {
            takeTaskCount.incrementAndGet();
        }
        LOG.log(Level.INFO, "\t take url: \t" + url);
        return url;
    }

    @Override
    public void offer(List<String> urls) {
        synchronized (lock) {
            for (String url : urls) {
                if (url != null && !bloomFilter.contains(url)) {
                    putTaskCount.incrementAndGet();
                    queue.offer(url);
                    bloomFilter.add(url);
                    LOG.log(Level.INFO, "\t offer url: \t" + url);
                }
            }
        }
    }

}
